package top.infopub.zookeeper.lock;


import java.util.concurrent.Semaphore;

import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.RevocationListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import top.infopub.exception.ServiceException;


/**
 * 基于ZK的分布式锁 </br>
 * 使用方式如下：<pre>
 *   /@Autowired 
 *   private CuratorFramework client;
 *   
 *   /@Transactional(rollbackFor = Exception.class)
 *   
 *   ZkLocker locker = new ZkLocker(client);
 *   try {
 *       boolean flag = locker.lock("repay/13888888888");
 *       if (flag) {
 *           // do something no transaction here……
 *           transactionalMethod();  // 锁在事务外，解决spring事务管理下并发更新问题 参考：https://my.oschina.net/u/3777556/blog/3011167?p=1
 *       }
 *   }
 *   catch (Exception e) {
 *       log.error("", e);
 *   }
 *   finally {
 *       locker.unlock();
 *   }
 *   
 *   /@Transactional(rollbackFor = Exception.class)
 *   public void transactionalMethod() {
 *       // do something transactional here……
 *   }
 *   
 * </pre>  
 * @author Awoke
 * @version 2018年10月25日
 * @see ZkLocker
 * @since
 */
public class ZkLocker {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private InterProcessMutex lock;// 重入的,排他的.

    private LockHolder _holder;

    private CuratorFramework client;

    private ConnectionStateListener stateListener = new StateListener();

    private LockEvent lockEvent;

    private String lockPath;

    private Semaphore semaphore = new Semaphore(0);

    private RevocationListener<InterProcessMutex> revocationListener;

    private String base_lock_path = "/mutexlock";

    private String split_char = "/";

    /**
     * spring容器下，<br>可以通过如下代码：
     * code-- @Autowired  CuratorFramework client 注入
     * @param client
     */
    public ZkLocker(CuratorFramework client) {
        this.client = client;
        this.lockEvent = new LockEvent() {
            @Override
            public boolean lose() {
                logger.debug("Lose,shoud be waiting or stop workers!");
                return false;
            }

            @Override
            public void beforeRelease() {
                logger.debug("Lock would be released!");
            }

            @Override
            public void afterAquire() {
                logger.debug("Locked success,can be running workers");
            }
        };
        revocationListener = new RevocationListener<InterProcessMutex>() {
            @Override
            public void revocationRequested(InterProcessMutex forLock) {
                if (!forLock.isAcquiredInThisProcess()) {
                    return;
                }
                try {
                    lockEvent.beforeRelease();
                    // 只有当前线程才可以释放
                    // 其他线程,将会抛出一个错误.
                    forLock.release();
                }
                catch (Exception e) {
                    logger.error("", e);
                }
            }
        };
    }

    /**
     * 锁
     * @param path 请务必使用  锁业务码 + '/' + 资源唯一标识
     * @return 
     * @see
     */
    public boolean lock(String path) {
        if (!StringUtils.contains(path, split_char)) {
            throw new ServiceException("路径必须为：锁业务码 + '/' + 资源唯一标识");
        }
        if (!StringUtils.startsWith(path, split_char)) {
            path = split_char + path;
        }
        path = StringUtils.removeEnd(path, split_char);
        path = base_lock_path + path;

        this.lockPath = path;
        lock = createLock(client);
        try {
            if (_holder == null) {
                _holder = new LockHolder();
                _holder.setDaemon(true);
            }
            if (!_holder.isAlive()) {
                _holder.start();
            }
            semaphore.acquire();// 直到_holder正常运行
        }
        catch (Exception e) {
            logger.error("lock error", e);
            return false;
        }
        return true;
    }

    /**
     * 释放当前锁
     * @see
     */
    public void unlock() {
        try {
            if (_holder.isAlive()) {
                _holder.interrupt();
            }
            // 兼容性处理 解决部分CuratorFramework(节点创建成永久性节点 无法自动删除) 手动删除之 by Awoke
            if (client.checkExists().forPath(lockPath).getDataLength() == 0) {
                client.delete().forPath(lockPath);
            }
        }
        catch (org.apache.zookeeper.KeeperException.NotEmptyException nee) {
            // ingore 高并发下 同一个lockPath在手动delete时会出现此异常 可以忽略   这里再尝试下删除
            try {
                if (client.checkExists().forPath(lockPath).getDataLength() == 0) {
                    client.delete().forPath(lockPath);
                }
            }
            catch (Exception xe) {}
        }
        catch (Exception e) {
            logger.error("unlock error", e);
        }
    }
    
    private InterProcessMutex createLock(CuratorFramework client) {
        lock = new InterProcessMutex(client, lockPath);
        // 协同中断,如果其他线程/进程需要此锁中断时,调用此listener.
        lock.makeRevocable(revocationListener);
        client.getConnectionStateListenable().addListener(stateListener);
        return lock;
    }

    private class StateListener implements ConnectionStateListener {
        @Override
        public void stateChanged(CuratorFramework client, ConnectionState newState) {
            switch (newState) {
                case LOST:
                    // 一旦丢失链接,就意味着zk server端已经删除了锁数据
                    boolean rebuild = lockEvent.lose();
                    _holder.interrupt();// NUll
                    if (rebuild) {
                        lock(lockPath);
                    }
                    break;
                default:
                    logger.debug(newState.toString());
            }
        }
    }

    static interface LockEvent {

        public void afterAquire();

        public boolean lose();

        // 释放锁
        public void beforeRelease();
    }

    class LockHolder extends Thread {
        @Override
        public void run() {
            try {
                lock.acquire();
                semaphore.release();
                lockEvent.afterAquire();
                synchronized (lock) {
                    lock.wait();
                }
            }
            catch (InterruptedException e) {
                try {
                    lockEvent.beforeRelease();
                    lock.release();
                }
                catch (Exception ie) {
                    logger.error("", ie);
                }
            }
            catch (Exception e) {
                logger.error("", e);
            }
        }
    }

    /**
     * Test
     * @param args
     * @throws Exception 
     * @see
     */
    public static void main(String[] args)
        throws Exception {
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
        // fluent style
        CuratorFramework client = builder.connectString("192.168.0.13:2191") // FixedEnsembleProvider
        .sessionTimeoutMs(30000).connectionTimeoutMs(30000) // cant connect to one observer-instance
        .retryPolicy(new ExponentialBackoffRetry(1000, 3)) // auto reconnect policy
        //.namespace(namespace) // good method,you can specify one fix prefiex-path of all znode.
        .defaultData(null).build();
        client.start();

        ZkLocker distLock = new ZkLocker(client);
        distLock.lock("/mutexlock/repay/15951899194");

        Thread.sleep(6000);
        distLock.unlock();
        client.close();
    }

}